package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.Collection;
import java.util.Set;

/**
 * @author Felix
 * @date 2023/2/1
 */
public class DimSinkFunction implements SinkFunction<JSONObject> {
    //将流中的数据写到Phoenix对应的维度表中 {"tm_name":"FDA","sink_table":"dim_base_trademark","id":12}
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        //获取输出目的地
        String tableName = jsonObj.getString("sink_table");
        //{"tm_name":"FDA","id":12}
        jsonObj.remove("sink_table");
        String type = jsonObj.getString("type");
        jsonObj.remove("type");
        //拼接向Phoenix表中插入数据的upsert语句 upsert into 表空间.表 (a,b,c) values(aa,bb,cc)
        //获取所有的属性
        Set<String> columns = jsonObj.keySet();
        Collection<Object> values = jsonObj.values();
        String upsertSql = "upsert into " + GmallConfig.PHOENIX_SCHEMA
            + "." + tableName + " (" + StringUtils.join(columns, ",")
            + ") values('" + StringUtils.join(values, "','") + "')";
        System.out.println("在Phoenix执行的upsert语句:" + upsertSql);

        PhoenixUtil.executeSql(upsertSql);
        //如果对维度表数据进行了更新操作，需要将Redis中对应的缓存数据清除掉
        if("update".equals(type)){
            DimUtil.delCached(tableName,jsonObj.getString("id"));
        }
    }
}
